package org.example.redis;

import org.apache.flink.api.connector.source.*;
import org.apache.flink.core.io.SimpleVersionedSerializer;

public class MyRedisSource implements Source<String, RedisSplit, Void> {

    private final String host;
    private final int port;
    private final String redisKey;

    public MyRedisSource(String host, int port, String redisKey) {
        this.host = host;
        this.port = port;
        this.redisKey = redisKey;
    }

    @Override
    public SourceReader<String, RedisSplit> createReader(SourceReaderContext readerContext) {
        return new RedisSourceReader(host, port, redisKey, readerContext);
    }

    @Override
    public Boundedness getBoundedness() {
        return Boundedness.CONTINUOUS_UNBOUNDED;
    }

    @Override
    public SplitEnumerator<RedisSplit, Void> createEnumerator(SplitEnumeratorContext<RedisSplit> enumeratorContext) {
        return new RedisSplitEnumerator(enumeratorContext);
    }

    @Override
    public SplitEnumerator<RedisSplit, Void> restoreEnumerator(SplitEnumeratorContext<RedisSplit> enumeratorContext, Void checkpoint) {
        return new RedisSplitEnumerator(enumeratorContext);
    }

    @Override
    public SimpleVersionedSerializer<RedisSplit> getSplitSerializer() {
        return new SimpleVersionedSerializer<RedisSplit>() {
            @Override
            public int getVersion() {
                return 1;
            }

            @Override
            public byte[] serialize(RedisSplit split) {
                return split.splitId().getBytes();
            }

            @Override
            public RedisSplit deserialize(int version, byte[] serialized) {
                return new RedisSplit(new String(serialized));
            }
        };
    }

    @Override
    public SimpleVersionedSerializer<Void> getEnumeratorCheckpointSerializer() {
        return new SimpleVersionedSerializer<Void>() {
            @Override
            public int getVersion() {
                return 1;
            }

            @Override
            public byte[] serialize(Void checkpoint) {
                return new byte[0];
            }

            @Override
            public Void deserialize(int version, byte[] serialized) {
                return null;
            }
        };
    }
}
